package redisDB

import (
	"fmt"
	"github.com/go-redis/redis"
	"rpcSrv/configs"
	"strconv"
	"sync"
	"time"
)

var (
	RedisClient *redis.Client
	once        sync.Once
)

func initClient() {
	conf := configs.GetRedisConf()
	db, _ := strconv.Atoi(conf.DB)
	RedisClient = redis.NewClient(&redis.Options{
		Addr:     fmt.Sprintf("%s:%s", conf.Ip, conf.Port),
		Password: conf.Password,
		DB:       db,
	})
}

//心跳断线重连
func heartbeat() {
	tick := time.NewTicker(3 * time.Second)
	for {
		select {
		case <-tick.C:
			err := RedisClient.Ping().Err()
			if utils.CheckErr(err) {
				//重连
				time.Sleep(time.Second)
				initClient()
			}
		}
	}
}

func GetRedisClient() (*redis.Client, error) {
	var err error
	once.Do(func() {
		initClient()
		go heartbeat()
	})
	_, err = RedisClient.Ping().Result()
	return RedisClient, err
}

func RedisConnClose() {
	RedisClient.Close()
}

type queue struct {
	queueName string
	receiver  func(msg string) error //消费者处理函数
	Opt       QueueOptions           //主要用于并发限速
}
type QueueOptions struct {
	Async     bool          //true就是并发处理  false同步处理
	TimeLimit time.Duration //设置等待时间限制并发速度
	ChanLimit int           //使用管道限制并发数量
}

//初始化队列
func NewQueue(queueName string, rcv func(msg string) error, opt ...QueueOptions) *queue {
	GetRedisClient()
	q := new(queue)
	q.queueName = queueName
	q.receiver = rcv
	if len(opt) > 0 {
		q.Opt = opt[0]
	}
	go q.listen()
	return q
}

func (this *queue) Err() error {
	return RedisClient.Ping().Err()
}

func (this *queue) GetQueueName() string {
	return this.queueName
}

func (this *queue) listen() {
	var ch chan struct{}
	//如果使用配置信息使用chan限速那么就new出一个限速chan
	if this.Opt.ChanLimit > 0 {
		ch = make(chan struct{}, this.Opt.ChanLimit)
	}
	var (
		rcvFn = func(data string) {
			if err := this.receiver(data); utils.CheckErr(err) {
				//错误信息保存到数据库待后期恢复
				//errMsg := new(entitys.RMQErrMsg)
				//errMsg.Msg = data
				//errMsg.QueueName = this.queueName
				//errMsg.CreatedAt,_ = utils.TimeString2Time(utils.GetCurrentTime())
				//errMsg.Err = err.Error()
				//db, err := mysqlGORM.GetMysqlDB()
				//if err == nil {
				//	db.Table(errMsg.TableName()).Create(errMsg)
				//}
			}
			//如果使用chan限速，每次消费完毕那么就释放一个chan资源
			if ch != nil {
				go func() {
					<-ch
				}()
			}
		}
		async     bool
		timeLimit time.Duration
	)
	async = this.Opt.Async
	if this.Opt.TimeLimit > 0 {
		timeLimit = this.Opt.TimeLimit
	}

	for {
		time.Sleep(timeLimit)
		msg, err := RedisClient.BRPop(0, this.queueName).Result()
		if utils.CheckErr(err) {
			time.Sleep(time.Second)
			fmt.Printf("队列：%v重连中...\n", this.queueName)
			go this.listen()
			return
		}
		//如果使用chan限速，每次消费前都往chan中记录一条消费记录，只有等该次消费完毕才释放
		if ch != nil {
			ch <- struct{}{}
		}
		if async {
			go rcvFn(msg[1])
		} else {
			rcvFn(msg[1])
		}
	}
}

//往队列发送发送数据
func (this *queue) Send(msg string) error {
	return RedisClient.LPush(this.queueName, msg).Err()
}
